package com.ayu.shortlink.project.mq.producer;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import com.ayu.shortlink.project.mq.basics.base.BaseSendMessageExtendDTO;
import com.ayu.shortlink.project.mq.basics.base.MessageWrapper;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;

import java.util.Optional;
import java.util.UUID;

/**
 * 抽象公共发布组件
 */
@RequiredArgsConstructor
@Slf4j
public abstract class AbstractCommonSendProduceTemplate<T> {
    private final RocketMQTemplate rocketMQTemplate;

    protected abstract BaseSendMessageExtendDTO buildBaseSendMessageExtendDTO(T messageSendEvent);

    protected abstract Message<?> buildMessage(T messageSendEvent,BaseSendMessageExtendDTO requestParam);

    /**
     * 实际发送的方法(实时)
     */
    public SendResult sendMessage(T messageSendEvent){
        BaseSendMessageExtendDTO baseSendMessageExtendDTO = buildBaseSendMessageExtendDTO(messageSendEvent);
        SendResult sendResult;
        try {
            StringBuilder destinationBuilder = StrUtil.builder().append(baseSendMessageExtendDTO.getTopic());
            if (StrUtil.isNotBlank(baseSendMessageExtendDTO.getTag())) {
                destinationBuilder.append(":").append(baseSendMessageExtendDTO.getTag());
            }
            sendResult = rocketMQTemplate.syncSend(
                    destinationBuilder.toString(),
                    buildMessage(messageSendEvent, baseSendMessageExtendDTO),
                    baseSendMessageExtendDTO.getSentTimeout(),
                    Optional.ofNullable(baseSendMessageExtendDTO.getDelayLevel()).orElse(0)
            );
            log.info("[生产者] {} - 发送结果：{}，消息ID：{}，消息Keys：{}", baseSendMessageExtendDTO.getEventName(), sendResult.getSendStatus(), sendResult.getMsgId(), baseSendMessageExtendDTO.getKeys());
        }catch (Throwable throwable){
            log.error("[生产者] {} - 消息发送失败，消息体：{}", baseSendMessageExtendDTO.getEventName(), JSON.toJSONString(messageSendEvent), throwable);
            throw throwable;
        }
        return sendResult;
    }

    /**
     * 定时发送的方法(定时)
     */
    @SneakyThrows
    public SendResult sendMessage(T messageSendEvent, Long deliverTimeStamp){
        if(deliverTimeStamp == null || deliverTimeStamp <= 0L){
            return sendMessage(messageSendEvent);
        }
        BaseSendMessageExtendDTO baseSendMessageExtendDTO = buildBaseSendMessageExtendDTO(messageSendEvent);
        SendResult sendResult;
        try {
            String keys = StrUtil.isEmpty(baseSendMessageExtendDTO.getKeys()) ? UUID.randomUUID().toString() : baseSendMessageExtendDTO.getKeys();
            org.apache.rocketmq.common.message.Message message = new org.apache.rocketmq.common.message.Message();
            if (StrUtil.isNotBlank(baseSendMessageExtendDTO.getTag())) {
                message.setTags(baseSendMessageExtendDTO.getTag());
            }
            message.setBody(JSON.toJSONBytes(new MessageWrapper<>(baseSendMessageExtendDTO.getKeys(),baseSendMessageExtendDTO)));
            message.setTopic(baseSendMessageExtendDTO.getTopic());
            message.setKeys(keys);
            message.setDeliverTimeMs(deliverTimeStamp);
            DefaultMQProducer producer = rocketMQTemplate.getProducer();
            sendResult = producer.send(message);

            log.info("[生产者] {} - 发送结果：{}，消息ID：{}，消息Keys：{}", baseSendMessageExtendDTO.getEventName(), sendResult.getSendStatus(), sendResult.getMsgId(), baseSendMessageExtendDTO.getKeys());
        }catch (Throwable throwable){
            log.error("[生产者] {} - 消息发送失败，消息体：{}", baseSendMessageExtendDTO.getEventName(), JSON.toJSONString(messageSendEvent), throwable);
            throw throwable;
        }
        return sendResult;
    }
}
